package com.bungarus.busi;

import com.bungarus.Transfer;
import com.bungarus.cluster.NodeInfo;
import com.bungarus.codec.protobuf.MessageProtobuf;
import com.bungarus.model.*;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
import io.netty.handler.timeout.IdleStateHandler;
import org.apache.log4j.Logger;
import org.springframework.data.mongodb.core.MongoTemplate;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by tang on 2018/11/29.
 */
public class TransferInnerClient implements BaseClient{
    Logger logger = Logger.getLogger(TransferInnerClient.class);

    private static final long RECONNECT_INTERVAL = 5;
    private static final int MAX_RECONNECT_COUNT = 3;

    private static final int RETRY_INIT_COUNT = 1;
    private static final int RETRY_TOTAL_COUNT = 2;

    private static final String RETRY_FAILED_MESSAGE_COLLECTION = "retry-failed-message";

    private AtomicInteger reconnectCount = new AtomicInteger(1);

    private final NodeInfo node;
    private NioEventLoopGroup eventLoopGroup;

    private ConcurrentHashMap<Long, AtomicInteger> retryQueue = new ConcurrentHashMap<>();

    /**
     * netty channel peers to client
     */
    private Channel channel;
    /**
     * mongo operator for persisting the retry-failed message
     */
    private RetryFailedMessageRepository retryFailedMessageRepository;

    public TransferInnerClient(NodeInfo node, MongoTemplate mongoTemplate4FailedRetry) {
        this.node = node;
        retryFailedMessageRepository = new RetryFailedMessageRepository(mongoTemplate4FailedRetry);
    }

    @Override
    public Channel getChannel() {
        return channel;
    }

    public void run() throws BungarusClientException {
        run(null);
    }

    @Override
    public void run(String connectionFrom) throws BungarusClientException {
        connect(null, null);
    }

    @Override
    public void connect(AtomicInteger index, EventLoopGroup eventLoop) throws BungarusClientException {
        if(null == eventLoop) {
            this.eventLoopGroup = new NioEventLoopGroup();
        }
        else {
            this.eventLoopGroup.shutdownGracefully();
            this.eventLoopGroup = null;
            this.eventLoopGroup = new NioEventLoopGroup();
        }

        Bootstrap client = new Bootstrap();
        client.group(this.eventLoopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                .handler(new MessageChannelInitializer(this));
        try {
            logger.info("================client: transfer connects "
                    + node.getIp() + ":" + node.getMessagePorts().get(Transfer.TAG));
            ChannelFuture channelFuture = client.connect(node.getIp(), node.getMessagePorts().get(Transfer.TAG));
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (!channelFuture.isSuccess()) {
                        if(reconnectCount.getAndIncrement() <= MAX_RECONNECT_COUNT) {
                            channelFuture.channel().eventLoop().schedule(() -> {
                                try {
                                    connect(index, eventLoop);
                                }
                                catch (BungarusClientException e) {
                                    e.printStackTrace();
                                }
                            }, RECONNECT_INTERVAL, TimeUnit.SECONDS);
                        }
                        else {
                            reconnectCount.set(1);
                            throw new IllegalStateException("node in busi cluster is unavailable. ip: "
                                    + node.getIp() + " port: " + node.getMessagePorts().get(Transfer.TAG));
                        }
                    }
                }
            });
            channel = channelFuture.sync().channel();
        }
        catch (Exception e) {
            throw new BungarusClientException("node in busi cluster is unavailable. ip: "
                    + node.getIp() + " port: " + node.getMessagePorts().get(Transfer.TAG)
                    + " , " + e.getMessage());
        }
    }

    @Override
    public void connectNextNode(EventLoopGroup eventLoopGroup) throws BungarusClientException {
        try {
            connect(null, eventLoopGroup);
        }
        catch (Exception e){
            throw new BungarusClientException(e);
        }
    }

    @Override
    public void reLoadRemoteServers(EventLoopGroup eventLoopGroup) {
        throw new IllegalStateException("not support");
    }

    @Override
    public void sendMessage(MessageProtobuf.Message msg, String clusterTag, ChannelHandlerContext forAck) throws BungarusClientException {
        if(null == channel || !channel.isActive()) {
            throw new BungarusClientException("no channel can be used to send message.");
        }
        MessageProtobuf.Message newMsg ;
        if(msg.getHeader().getType() == MessageType.CLONE) {
            newMsg = msg;
        }
        else {
            newMsg = MessageBuilder.cloneMessageWithNewId(msg);
        }

        channel.writeAndFlush(newMsg).addListener(channelFuture -> {
            //after sending message, the message is putted into a queue to wait for being acknowledged
            if(channelFuture.isDone() && channelFuture.isSuccess()) {
                if(null != forAck) {
                    forAck.writeAndFlush(MessageBuilder.buildAckMessage(msg.getHeader().getId()));
                }
                DischargeQueue dischargeQueue = Discharger.getQueueInstance();
                dischargeQueue.addToQueue(newMsg);
            }
            else {//retry
                Long id = Long.valueOf(newMsg.getHeader().getId());
                int e;
                if(null == retryQueue.get(id)){
                    AtomicInteger ai = new AtomicInteger(RETRY_INIT_COUNT);
                    e = ai.getAndIncrement();
                    retryQueue.put(id, ai);
                }
                else {
                    e = retryQueue.get(id).getAndIncrement();
                    if(e > RETRY_TOTAL_COUNT) {
                        //save the message into persistent storage (supported by mongo)
                        saveTryFailedMessage(newMsg);
                        ///////////////////////////////////////////////////////////
                        try {
                            channel.close();
                            this.connectNextNode(this.eventLoopGroup);
                        }
                        catch (BungarusClientException e1){
                            e1.printStackTrace();
                        }
                        finally {
                            retryQueue.remove(id);
                        }
                        return;
                    }
                }
                Thread.sleep((long)Math.pow(2, e) * 1000);
                sendMessage(newMsg, clusterTag, forAck);
            }
        });
    }

    @Override
    public void saveTryFailedMessage(MessageProtobuf.Message failedMsg) {
        RetryFailedMessage.Builder builder = RetryFailedMessage.newBuilder();
        builder.id(Long.valueOf(failedMsg.getHeader().getId()))
                .app(failedMsg.getHeader().getApp())
                .clusterTag(Transfer.TAG)
                .content(failedMsg.getBody().getContent())
                .dev(failedMsg.getHeader().getDev())
                .from(failedMsg.getHeader().getFrom())
                .gaction(failedMsg.getHeader().getGaction())
                .lang(failedMsg.getHeader().getLang())
                .priority(failedMsg.getHeader().getPriority())
                .sentTime(failedMsg.getHeader().getSentTime())
                .tenant(failedMsg.getHeader().getTenantID())
                .to(failedMsg.getHeader().getTo())
                .type(failedMsg.getHeader().getType());

        retryFailedMessageRepository.add(builder.built(), RETRY_FAILED_MESSAGE_COLLECTION);
    }

    private static class MessageChannelInitializer extends ChannelInitializer<SocketChannel> {
        private static final int ALL_IDLE_TIMEOUT = 5;

        private final BaseClient client;

        public MessageChannelInitializer(BaseClient client) {
            this.client = client;
        }

        @Override
        protected void initChannel(SocketChannel socketChannel) throws Exception {
            socketChannel.pipeline().addLast(new IdleStateHandler(0,
                    0, ALL_IDLE_TIMEOUT, TimeUnit.SECONDS));
            socketChannel.pipeline().addLast(new ProtobufVarint32FrameDecoder());
            socketChannel.pipeline().addLast(new ProtobufDecoder(MessageProtobuf.Message.getDefaultInstance()));
            socketChannel.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
            socketChannel.pipeline().addLast(new ProtobufEncoder());
            socketChannel.pipeline().addLast(new NodeAsClientMessageHandler());
            socketChannel.pipeline().addLast(new TransferClientHeartbeatHandler(client));
        }
    }
}
